package station;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.MulticastSocket;
import java.net.NetworkInterface;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Deque;
import java.util.LinkedList;
import java.util.Random;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import util.DataLogger;
import util.DataPackage;

/**
 * Station
 * implements the TDMA algorithm with up to 20 Stations.
 * Waits for Data from a Source and sends the data over Multicast to the other Stations.
 * Waits for incoming Data over Multicast and sends it to a Sink. 
 * 
 * @author Jan Quenzel
 * @author Rico Flaegel
 */
public class Station implements Runnable {
	/**
	 * Standard constructor
	 * @param address Multicast address
	 * @param mcastport MulticastPort
	 * @param receiveport ReceivePort for incoming Data from Source
	 * @param sendport SendPort for outgoing data to Sink
	 */
	public Station(String address, int mcastport, int receiveport, int sendport){
		this(address, mcastport, null ,receiveport, sendport);
	}
	/**
	 * Standard constructor
	 * @param address Multicast address
	 * @param mcastport MulticastPort
	 * @param interfaceName NetworkInterface
	 * @param receiveport ReceivePort for incoming Data from Source
	 * @param sendport SendPort for outgoing data to Sink
	 */
	public Station(String address, int mcastport, String interfaceName,int receiveport, int sendport){
		this.mcastPort = mcastport;
		this.sinkPort = sendport;
		this.receiveAndSendPort = receiveport;
		this.address = address;
		this.networkInterfaceName = interfaceName;
	}
	/**
	 * Multicast Port
	 */
	private int mcastPort;
	/**
	 * Port of the Sink
	 */
	private int sinkPort;
	/**
	 * port where data from Source will be received
	 */
	private int receiveAndSendPort;
	/**
	 * Multicast address
	 */
	private String address;
	/**
	 * address where to receive from station. standard = localhost
	 */
	private String receiveAddress = "localhost";
	/**
	 * Name of the Network Interface used to send for Multicast
	 */
	private String networkInterfaceName;
	/**
	 * Socket used to receive from Source and send to Sink
	 */
	private DatagramSocket receiveAndSendSock;
	/**
	 * Multicast Socket used to Send
	 */
	private MulticastSocket mCastSendSock;
	/**
	 * Multicast Socket used to Receive
	 */
	private MulticastSocket mCastRecSock;
	/**
	 * DatagramPacket used to Send
	 */
	private DatagramPacket sdp;
	/**
	 * Multicast Group address
	 */
	private InetAddress group;
	/**
	 * StationReceiver, receiving from Source
	 */
	private StationReceiver myReceiver;
	/**
	 * StationMulticastReceiver, receiving DataPackages from Stations via Multicast
	 */
	private StationMultiCastReceiver myMcastReceiver;
	/**
	 * StationScheduler, creates the Schedule and determines when to send
	 */
	private StationScheduler sched;
	/**
	 * actual frame
	 */
	private Frame act;
	/**
	 * next, upcoming frame
	 */
	private Frame next;
	/**
	 * RNG
	 */
	private Random rnd;
	/**
	 * Timer used to schedule TimerTasks to send the Package inside the Slot
	 */
	private Timer tm;
	/**
	 * Deque as an inputbuffer for data from Source
	 */
	private Deque<ByteBuffer> sendQueue;
	/**
	 * Datalogger to log messages and what happens
	 */
	private DataLogger logger;
	/**
	 * DataPackage used to send
	 */
	private DataPackage sendDP;
	
	/**
	 * number of slots
	 */
	private static int slotsize = 20;
	/**
	 * delay from Slot beginning
	 */
	private int delay = 25;
	/**
	 * duration of a frame (1 sec.)
	 */
	private int timerPeriod = 1000;
	/**
	 * time for one slot
	 */
	private int slotTime = 50;
	/**
	 * Number of Frames the station won't send
	 */
	private int backoff = 0;
	/**
	 * maximum backoff
	 */
	private int maxBackOff = 5;
	/**
	 * mySlot where i want to send
	 */
	private int mySlot = -1;
	/**
	 * length of deque
	 */
	private int queueLen = 10;
	private long lastSendTime = 0;
	
	private boolean running = true;
	/**
	 * @param args <address> <mcastport> [<networkinterface>] <sourceport> <sinkport>
	 */
	public static void main(String[] args) {
		try{
			int mcastport = 0;
			int sourceport = 0;
			int sinkport = 0;
			Station sr	= null;
			if(args.length == 4){
				mcastport = Integer.parseInt(args[1]);
				sourceport = Integer.parseInt(args[2]);
				sinkport = Integer.parseInt(args[3]);
				sr = new Station(args[0],mcastport,sourceport,sinkport);
			}else if(args.length == 5){
				mcastport = Integer.parseInt(args[1]);
				sourceport = Integer.parseInt(args[3]);
				sinkport = Integer.parseInt(args[4]);
				sr = new Station(args[0],mcastport,args[2],sourceport,sinkport);		
			}else{
				throw new NumberFormatException();
			}
			if(sr != null){
				ExecutorService exec = Executors.newSingleThreadExecutor();
				exec.execute(sr);
				exec.shutdown();
			}
		}catch(NumberFormatException e){
			System.err.println("Wrong number of Input parameters... Usage: Sender_Receiver <address> <mcastport> [<networkinterface>]" +
					" <sourceport> <sinkport>");
		}
	}
	
	@Override
	public void run() {
		try {
			initStation();
			callTimer();
		} catch (UnknownHostException e) {
			e.printStackTrace();
		} catch (IOException e1) {
			e1.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	/**
	 * initialize the Station.
	 * @throws UnknownHostException
	 * @throws IOException
	 */
	public void initStation() throws UnknownHostException, IOException{
		this.logger = new DataLogger((receiveAndSendPort+"LogFile_SendRec.txt"));
		System.out.println("logger created");
		this.mCastSendSock = new MulticastSocket(mcastPort);
		this.mCastRecSock = new MulticastSocket(mcastPort);
		this.receiveAndSendSock = new DatagramSocket(receiveAndSendPort);
		this.group = InetAddress.getByName(address);
		if(networkInterfaceName != null){
			this.mCastSendSock.joinGroup(new InetSocketAddress(address,mcastPort),NetworkInterface.getByName(networkInterfaceName));
			this.mCastRecSock.joinGroup(new InetSocketAddress(address,mcastPort),NetworkInterface.getByName(networkInterfaceName));
		}else{
			this.mCastSendSock.joinGroup(group);
			this.mCastRecSock.joinGroup(group);
		}
		this.myMcastReceiver = new StationMultiCastReceiver(mCastRecSock,receiveAndSendSock);
		this.myReceiver = new StationReceiver(receiveAndSendSock);
		this.sendQueue = new LinkedList<ByteBuffer>();
		this.logger.logString("Sockets created\n");
		this.sendDP = null;
		this.act = new Frame();
		this.next = new Frame();
		this.rnd = new Random();
		this.tm = new Timer();
		this.act.frameStartTime = 0;
		this.sched = new StationScheduler(this);
		this.logger.logString("DataSourceReceiver + MultiCastReceiver started\n");
		byte[] sb = new byte[DataPackage.len];
		this.sdp = new DatagramPacket(sb, sb.length, group, mcastPort);
		ExecutorService exec = Executors.newFixedThreadPool(2);
		exec.execute(this.myMcastReceiver);
		exec.execute(this.myReceiver);
		exec.shutdown();
		backoff = 1;
	}
	
	/**
	 * Build the Schedule every beginning of a frame.
	 * @throws InterruptedException
	 */
	public void callTimer() throws InterruptedException {
		long sleep = System.currentTimeMillis();
		sleep = timerPeriod - sleep%timerPeriod-2;
		boolean sle=true;
		System.out.println(sleep);
		// init
		TimeUnit.MILLISECONDS.sleep(timerPeriod + sleep);
		// start
		while (running) {
			while(VirtualTime.getTime()<=next.frameStartTime);
			sched.doJob();
			if(sle){ sleep = timerPeriod; sle = false; }
			else sleep =((VirtualTime.getTime()/timerPeriod)*timerPeriod+timerPeriod)-VirtualTime.getTime();// next.frameStartTime-VirtualTime.getTime();
			//System.out.println("sleep: "+sleep);
			TimeUnit.MILLISECONDS.sleep(sleep);
		}
	}
	/**
	 * Send the Message
	 * @return sendMessage successfully?
	 * @throws IOException
	 */
	public boolean sendMessage() throws IOException{
		synchronized(sendQueue){
			if(sendQueue.peek()!=null && mySlot != -1){
				DataPackage DP = new DataPackage();
				DP.setData(sendQueue.remove().array());
				DP.setNextSlotNumber((byte) act.mySlot);
				DP.setTime( System.currentTimeMillis() );//DP.setTime(act.correctSendTime);
				sdp.setData(DP.toByte());
				this.lastSendTime = DP.getTime();
				sendDP = DP;
				mCastSendSock.send(sdp);
				return true;
			}
		}
		return false;
	}
	
	/**
	 * Receiver of the Data from the Source.
	 * Data will be stored in a Deque used as an Inputbuffer.
	 * 
	 * @author Jan Quenzel
	 * @author Rico Flaegel
	 */
	private class StationReceiver implements Runnable{
		/**
		 * Standard Constructor for StationReceiver
		 * @param receiveSock Socket on which data will be received
		 */
		public StationReceiver(DatagramSocket receiveSock) {
			this.receive = receiveSock;
		}
		/**
		 * DatagramPacket to receive
		 */
		private DatagramPacket rdp;
		/**
		 * DatagramSocket where data will be received.
		 */
		private DatagramSocket receive;
		
		@Override
		public void run() {
			this.rdp = new DatagramPacket(new byte[DataPackage.len],DataPackage.len);
			try {
				while(running){
					receive.receive(rdp);
					synchronized(sendQueue){
						while(sendQueue.size()> queueLen && queueLen>1){ // remove old data from queue
							sendQueue.removeFirst();
						}
						sendQueue.add(ByteBuffer.wrap(rdp.getData().clone()));
					}
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	/**
	 * StationMultiCastReceiver
	 * Receives Multicast messages from Stations, processes the messages and forwards to the Sink.
	 * 
	 * @author Jan Quenzel
	 * @author Rico Flaegel
	 */
	private class StationMultiCastReceiver implements Runnable{
		/**
		 * standard Constructor for StationMulticastReceiver
		 * @param rec MulticastSocket to receive from Stations
		 * @param send Socket to forward to Sink.
		 */
		StationMultiCastReceiver(MulticastSocket rec, DatagramSocket send){
			this.receive = rec;
			this.send = send;
		}
		/**
		 * DatagramPacket to send
		 */
		private DatagramPacket sdp;
		/**
		 * DatagramPacket to receive
		 */
		private DatagramPacket rdp;
		/**
		 * DatagramSocket to forward to Sink
		 */
		private DatagramSocket send;
		/**
		 * MulticastSocket to receive from Stations
		 */
		private MulticastSocket receive;
		
		@Override
		public void run() {
			try {
				byte[] sb = new byte[DataPackage.len];
				byte[] rb = new byte[DataPackage.len];
				DataPackage dp = new DataPackage();
				sdp = new DatagramPacket(sb, sb.length, InetAddress.getByName(receiveAddress), sinkPort);
				rdp = new DatagramPacket(rb, rb.length);
				// init & wait for border of second
				try {
					long sleep = VirtualTime.getTime();
					sleep = timerPeriod - sleep%timerPeriod-2;
					TimeUnit.MILLISECONDS.sleep(sleep);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				while(running){
					receive.receive(rdp);
					//send first
					sdp.setData(rdp.getData(), 0, rdp.getData().length);
					send.send(sdp); // send to Sink
					// then do other things...
					long timestamp = VirtualTime.getTime();
					long ts = System.currentTimeMillis();
					dp.fromByte(sdp.getData());
//					System.out.println("send: "+dp.getTime()+" rectime: "+timestamp+" ts:"+ts+" dif:"+(timestamp-dp.getTime())+" vtd:"+VirtualTime.diff);
					sched.orientOnOtherID(dp, ts,dp.equals(sendDP));
					if(!dp.equals(sendDP) && timestamp >= (act.mySendTime-delay) && timestamp < (act.mySendTime + delay)){
						//collision detected
						System.out.println("collision!");
						backoff = 1;//rnd.nextInt(maxBackOff)+1;
					}
					// end
				}
			} catch (IOException e) {
				e.printStackTrace();
			} finally{
				receive.close();
				send.close();
			}
		}	
	}
	/**
	 * StationMulticastSender
	 * TimerTask to schedule the sending of a message.
	 * @author Jan Quenzel
	 * @author Rico Flaegel
	 */
	private class StationMultiCastSender extends TimerTask{
		public void run() { // called by Timer
			try {
				sendMessage(); // sendMessage
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	/**
	 * StationScheduler
	 * builds the Schedules and initializes the sending process, marks incoming messages in Frame
	 * 
	 * @author Jan Quenzel
	 * @author Rico Flaegel
	 */
	private class StationScheduler{
		/**
		 * this Station, which the Scheduler is part of
		 */
		private Station station;
		/**
		 * am i now ready again after a collision?
		 */
		private boolean againAfterCollision;
		/**
		 * slot where to send
		 */
		private int number;
		
		/**
		 * Standard Constructor for the Station Scheduler
		 * @param st the Station
		 */
		StationScheduler(Station st){
			this.station = st;
			againAfterCollision = false;
			number = -1; // none selected
		}
		/**
		 * Marks used slots inside the next frame. Watches out for our slot being marked by someone else.
		 * @param dp received DataPackage
		 * @param ts timestamp
		 * @param equals package is equals to which station send
		 */
		synchronized void orientOnOtherID(DataPackage dp, long ts,boolean equals){
			VirtualTime.addToDiff(System.currentTimeMillis()-dp.getTime()); //calculate difference
			if(equals){
				if(next.slots[dp.getNextSlotNumber()%next.slots.length] == true){ //collision, need to get new slot
					number = -1;
				}else{
					next.slots[dp.getNextSlotNumber()%next.slots.length] = true; //my package, set number to own slot
					number = dp.getNextSlotNumber()%next.slots.length;
				}
			}else{
				if((dp.getNextSlotNumber()%next.slots.length) == number) { 
					number = -1;  // some one else wants that slot, we look for another slot
				}
				next.slots[dp.getNextSlotNumber()%next.slots.length] = true; //mark slot
			}
		}
		/**
		 * build the Schedule for the next Frame.
		 * will be called at the beginning of the next Frame.
		 * calculates the send time for our Message/Package
		 */
		synchronized void buildSchedule(){
			next.frameStartTime = (VirtualTime.getTime()/timerPeriod)*timerPeriod;
			if(backoff == 0){ // we want to send
				mySlot = act.mySlot;
				next.mySlot = next.getNextFreeSlot(number,againAfterCollision);
//				number = -1;
				againAfterCollision = false;
				if(next.mySlot != -1) next.slots[next.mySlot]=true; // we got a slot, set our true
				next.mySendTime = next.frameStartTime + next.mySlot*slotTime+delay;
				next.correctSendTime = next.frameStartTime + next.mySlot*slotTime+delay;
			}else{
				backoff--;
				next.mySendTime = 0;
				againAfterCollision = true;
				number = -1;
			}
			act = next; // swap frames, next becomes actual frame.
			next = new Frame(); // create another frame
			next.frameStartTime = (VirtualTime.getTime()/timerPeriod)*timerPeriod + timerPeriod;
			String log = " actual frame startTime: "+act.frameStartTime; // do some logging
			if( next.mySendTime > 0 ) log += (" actual Send Time:" + act.mySendTime + " rt:"+ System.currentTimeMillis());
			log += (" "+next.frameStartTime+ " Delta: ");
			logger.logString(log);
			System.out.println(log);
		}
		/**
		 * lets the StationScheduler do his Job, build Schedule and schedule the sending of the message
		 */
		public void doJob(){
			buildSchedule();
			String log = "";
			if(act.mySendTime > 0 && act.mySendTime - VirtualTime.getTime() > 0){ // schedule the sending
				tm.schedule(new StationMultiCastSender(), act.mySendTime - VirtualTime.getTime() );
				log = "Scheduled to send at "+act.mySendTime;
			}else{// error couldn't send
				number = -1;
				log = "Couldn't schedule to send. Time lies in past.";
				System.out.println("couldn't schedule! Slot: " + act.mySlot+" time: "+(act.mySendTime - VirtualTime.getTime())+" st: "+act.mySendTime+ " vt: "+VirtualTime.getTime());
			}
			logger.logString(log);
		}
	}
	/**
	 * Frame
	 * contains field containing the slots, the startTime and when we want to sent in this frame.
	 * 
	 * @author Jan Quenzel
	 * @author Rico Flaegel
	 */
	private class Frame{
		/**
		 * field with slots
		 * false = free, true = marked as used
		 */
		boolean [] slots;
		/**
		 * my sending slot
		 */
		int mySlot = -1;
		/**
		 * start time of this frame
		 */
		long frameStartTime = 0;
		/**
		 * scheduled time to send (0 = not planned to send anything)
		 */
		long mySendTime = 0;
		/**
		 * corrected Send Time 
		 * @deprecated
		 */
		long correctSendTime = 0;
		
		/**
		 * Standard Constructor for a Frame
		 */
		public Frame(){
			this.slots = new boolean[slotsize];
			for(int i = 0; i < slots.length; i++) slots[i] = false;
		}
		/**
		 * calculates the free Slot in Next Frame, where we plan to send
		 * @param last last slot number, -1 equals an error or not being able to send in previous frame
		 * @param againAfterCollision searching after Collision again, currently automatic
		 * @return
		 */
		private int getNextFreeSlot(int last, boolean againAfterCollision){
			if(last == -1){
				ArrayList<Integer> al = new ArrayList<Integer>(20); // get all free slots, save in list
				for (int i = 0; i < slots.length; i++){
					if (!slots[i]){
						al.add(i);
					}
				}
				if (al.size() == 0){ // no free slot available
					return -1;
				}else{
//					if(/*againAfterCollision && */(al.size() > 3)){
						return al.get(rnd.nextInt(al.size())); // compute random free slot
//					}else{
//						return al.get(0);
//					}
				}
			}else{
				return last; // use old slot
			}
		}
	}
	/**
	 * VirtualTime
	 * represents the virtual time ( real time - diff )
	 * methods to add time diffs and calculate the difference and get the current virtual time
	 * 
	 * @author Jan Quenzel
	 * @author Rico Flaegel
	 */
	private static class VirtualTime{
		/**
		 * ring buffer array containing "Number of Slots" differences between receive time and send time of a package
		 */
		public static long [] deltaArray = new long[slotsize];
		/**
		 * difference over all 20 entries in deltaArray
		 */
		public static long diff = 0;
		/**
		 * Pointer to build a ring buffer, actual position will get the next value
		 */
		public static int ringbufferPointer = 0;
		/**
		 * adds difference to the virtual Time and calculates the new diff
		 * @param timer difference between received real time and send real time
		 */
		public static synchronized void addToDiff(long timer){
			deltaArray [ringbufferPointer] = timer;
			ringbufferPointer = (ringbufferPointer+1)%slotsize;
			calcDiff();
		}
		/**
		 * calculates diff over all elements inside the ring buffer
		 * gets smoothed by 0 elements on startup. 
		 * (when only one message received, 0 sized differences will be taken in account too, making starting diff smoother) 
		 */
		public static synchronized void calcDiff(){
			long delta = 0;
			for(int i = 0; i < deltaArray.length; i++){
				delta += deltaArray[i];
			}
			diff = delta / deltaArray.length;
		}
		
		/**
		 * Gets the current virtual time.
		 * @return virtual Time
		 */
		public static long getTime(){
			return System.currentTimeMillis() - diff;
		}
	}
}
